# # @Version        : 1.0
# # @Update Time    : 2024/12/18 23:48
# # @File           : sqlalchemy_event.py
# # @IDE            : PyCharm
# # @Desc           : SQLAlchemy事件监听器，用于处理数据库变更事件

from typing import Dict, List, Any, Optional
from sqlalchemy import event, text
from sqlalchemy.orm import Mapper
from sqlalchemy.engine.base import Connection
from src.log import logger
from src.models import SysDictType, SysDictData, SysRole
from src.services.cache_service import CacheService
import asyncio
from functools import wraps
import time


def async_task_handler(func):
    """异步任务装饰器，用于处理异步操作和错误"""

    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError as e:
            logger.error("获取事件循环失败: %s", e)
            return

        try:
            coro = func(*args, **kwargs)
            future = asyncio.run_coroutine_threadsafe(coro, loop)
            future.add_done_callback(handle_future_result)
        except Exception as e:
            logger.error("异步任务创建失败: %s", e, exc_info=True)

    return wrapper


def handle_future_result(future):
    """处理异步任务结果"""
    try:
        future.result()
    except Exception as e:
        logger.error("异步任务执行失败: %s", e, exc_info=True)


def query(connection: Connection) -> Dict[str, List[Dict[str, Any]]]:
    """同步查询数据库，获取字典数据

    Args:
        connection: 数据库连接对象

    Returns:
        Dict[str, List[Dict[str, Any]]]: 包含字典类型和字典数据的字典
    """
    start_time = time.time()
    try:
        # 查询字典类型
        type_result = connection.execute(text("SELECT * FROM sys_dict_type"))
        dict_types = [dict(row) for row in type_result.mappings()]

        # 查询字典数据
        data_result = connection.execute(
            text(
                "SELECT id, dict_sort, dict_label, dict_value, "
                "dict_type_id, is_default, status, remark FROM sys_dict_data"
            )
        )
        dict_data = [dict(row) for row in data_result.mappings()]

        logger.debug("数据库查询耗时: %.2f秒", time.time() - start_time)
        return {"dict_type": dict_types, "dict_data": dict_data}
    except Exception as e:
        logger.error(f"数据库查询失败: {str(e)}")
        raise


@async_task_handler
async def update_dict_cache(data: Dict[str, List[Dict[str, Any]]]):
    """更新字典缓存

    Args:
        data: 包含字典类型和字典数据的字典
    """
    try:
        await CacheService.get_dict_data_to_cache(data)
        logger.info("字典缓存更新成功")
    except Exception as e:
        logger.error("字典缓存更新失败: %s", e, exc_info=True)
        raise


def change_before(mapper: Mapper, connection: Connection, target):
    """数据修改事件处理器 - 同步监听器触发异步缓存更新

    Args:
        mapper: SQLAlchemy映射器
        connection: 数据库连接
        target: 目标对象
    """
    logger.info("检测到字典数据变更，准备刷新缓存")

    try:
        # 同步查询数据库获取最新数据
        data = query(connection)
        # 异步更新缓存
        update_dict_cache(data)
    except Exception as e:
        logger.error("字典数据变更处理失败: %s", e, exc_info=True)


def register_sqlalchemy_events():
    """注册SQLAlchemy事件监听器"""
    logger.info("注册ORM事件监听器")

    # 字典数据变更监听
    event.listen(SysDictType, "after_update", change_before)
    event.listen(SysDictType, "after_delete", change_before)
    event.listen(SysDictData, "after_insert", change_before)
    event.listen(SysDictData, "after_update", change_before)
    event.listen(SysDictData, "after_delete", change_before)


def remove_sqlalchemy_events():
    """注销所有事件监听器"""
    logger.info("注销ORM事件监听器")

    # 注销字典数据变更监听
    event.remove(SysDictType, "after_update", change_before)
    event.remove(SysDictType, "after_delete", change_before)
    event.remove(SysDictData, "after_insert", change_before)
    event.remove(SysDictData, "after_update", change_before)
    event.remove(SysDictData, "after_delete", change_before)
